import com.bff.gaia.api.common.functions.FlatMapFunction;
import com.bff.gaia.api.java.DataSet;
import com.bff.gaia.api.java.ExecutionEnvironment;
import com.bff.gaia.api.java.tuple.Tuple2;
import com.bff.gaia.util.Collector;

public class BatchWordCount {
    public static void main(String[] args) throws Exception {
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(5);

		String filePath;
		if (args.length > 0) {
			filePath = args[0];
		}
		else {
			// filePath = "file:///home/gaia/hfy/workspace/test/data.txt";
			filePath = "hdfs://master:9000/a1b1c1d1e1f1g1h1i1j1_128M";
		}
		DataSet<String> text = env.readTextFile(filePath);

        DataSet<Tuple2<String, Integer>> wordCounts = text
                .flatMap(new LineSplitter())
				.partitionByRange(0)
				// .setParallelism(3)
                .groupBy(0)
                .sum(1);

        wordCounts.print();
    }

    public static class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String line, Collector<Tuple2<String, Integer>> out) {
            for (String word : line.split(" ")) {
                out.collect(new Tuple2<String, Integer>(word, 1));
            }
        }
    }
}